package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo20Partition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo20Partition")
    conf.setMaster("local")

    // 设置Spark程序默认的并行度
    conf.set("spark.default.parallelism", "4")

    val sc: SparkContext = new SparkContext(conf)

    // 读取words单词数据

    // 第一个RDD的分区数量由切片数量决定
    val lineRDD: RDD[String] = sc.textFile("Spark/data/words")

    println("lineRDD的分区数量为：" + lineRDD.getNumPartitions)

    /**
     * 手动改变RDD的分区数量
     * repartition coalesce
     * repartition(numPartitions) = coalesce(numPartitions,shuffle = true)
     * 如果要增大分区：必须通过shuffle => repartition
     * 如果要减少分区：是可以不需要shuffle的 => coalesce
     */
    val reParRDD: RDD[String] = lineRDD.repartition(5)
    println("经过repartition过后的分区数:" + reParRDD.getNumPartitions)


    val coalesceRDD: RDD[String] = lineRDD.coalesce(10, shuffle = true)
    println("经过coalesce过后的分区数:" + coalesceRDD.getNumPartitions)

    val coalesceRDD2: RDD[String] = lineRDD.coalesce(1)
    println("第二个经过coalesce过后的分区数:" + coalesceRDD2.getNumPartitions)


    // Word Count
    // flatMap算子不会产生Shuffle
    // 子RDD的分区数默认等于父RDD的分区数量
    val wordsRDD: RDD[String] = lineRDD.flatMap(line => line.split(","))
    println("wordsRDD的分区数量为：" + wordsRDD.getNumPartitions)

    // map算子也不会产生Shuffle 分区数默认等于父RDD的分区数量
    val wordKVRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
    println("wordKVRDD的分区数量为：" + wordKVRDD.getNumPartitions)


    /**
     * reduceByKey算子是Shuffle类的算子 会产生Shuffle
     * 分区数量由以下几个因素决定：
     * 1、什么都不指定默认也等于父RDD的分区数量
     * 2、shuffle类的算子可以额外接收一个参数：numPartitions(手动指定分区数量，改变任务的并行度)
     * 3、通过一个Spark的参数(spark.default.parallelism)指定
     *
     * 优先级：
     * 额外接收的参数：numPartitions > 通过Spark的参数(spark.default.parallelism)指定 > 默认等于父RDD的分区数量
     *
     *
     */
    val wordCntRDD: RDD[(String, Int)] = wordKVRDD.reduceByKey(_ + _, 6)
    println("wordCntRDD的分区数量为：" + wordCntRDD.getNumPartitions)

    wordCntRDD.foreach(println)


  }

}
